package com.zhuangjie.mvc.sse.service;

import com.zhaungjie.base.utils.json.JsonUtils;
import com.zhuangjie.event.annotation.ModuleEventName;
import com.zhuangjie.event.service.AbstractModuleEventService;
import com.zhuangjie.mvc.sse.entity.SSEResult;
import lombok.Data;
import org.springframework.data.redis.connection.Message;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
/**
 * 事件数据
 *
 * @author zhuangjie
 * @date 2023/12/07
 */
@Data
class EventData<K>  {
    K targetId;
    SSEResult data;
}
@ModuleEventName(moduleNamePrefix = true)
public abstract class AbstractSSEService<K> extends AbstractModuleEventService implements SSEService<K>   {
    protected Map<K, SseEmitter> targets = new HashMap<>();

    public synchronized SseEmitter listen(K targetId) throws IOException {
        SseEmitter sseEmitter = targets.get(targetId);
        if (sseEmitter != null) remove(targetId, sseEmitter);
        // 0L表示一直监听，直到用户手动关闭
        sseEmitter = new SseEmitter(0L);
        targets.put(targetId, sseEmitter);
        // 向前端向送SSE open事件
        sseEmitter.send(SseEmitter.event().name("open-event").data("SSE连接成功！"));
        return sseEmitter;
    }

    public synchronized void remove(K targetId) {
        remove(targetId, find(targetId));
    }

    public synchronized void remove(K targetId, SseEmitter sseEmitter) {
        if (sseEmitter == null) return;
        sseEmitter.complete();
        targets.remove(targetId);
    }

    private SseEmitter find(K targetId) {
        return targets.get(targetId);
    }

    @Override
    public void send(K targetId, SSEResult sseResult) throws IOException {
        EventData<K> eventData = new EventData<>();
        eventData.setData(sseResult);
        eventData.setTargetId(targetId);
        emit(eventData);
    }


    /**
     * 开始维护
     */
    @Override
    public void startMaintenance() {
        CompletableFuture.runAsync(()->{
            for (Map.Entry<K, SseEmitter> entry : targets.entrySet()) {
                try {
                    entry.getValue().send(SseEmitter.event().comment("-heartbeat-"));
                } catch (Exception e) {
                    remove(entry.getKey());
                }
            }
        });
    }

    /**
     * 收到委托
     * @param message
     * @param pattern
     */
    @Override
    public void on(Message message, byte[] pattern) {
        String eventDataJson = new String(message.getBody(), StandardCharsets.UTF_8);
        EventData<K> eventData = JsonUtils.parse(eventDataJson, EventData.class);
        SseEmitter sseEmitter = find(eventData.targetId);
        try {
            if (sseEmitter == null)  {
                return;
            }
            sseEmitter.send(eventData.data);
        }catch (Exception e) {
            // 当出现异常，说明信息无法改送过去，将该sseEmitter移除
            sseEmitter.complete();
            targets.remove(eventData.targetId);
        }
    }


}
